
/***************************************************************************
 *   sakcab 
 *                                                                         *
 *   Copyright (C) 2010 by kuerant                                         *
 *                         <kuerant@gmail.com>                             *
 *                                                                         *
 *   This library is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU Library General Public License as       *
 *   published by the Free Software Foundation; either version 2 of the    *
 *   License, or (at your option) any later version.                       *
 *                                                                         *
 *   This library is distributed in the hope that it will be usefull, but  *
 *   WITHOUT ANY WARRANTY; without even the implied warranty of            *
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU      *
 *   Library General Public License for more details.                      *
 *                                                                         *
 *   You should have received a copy of the GNU Library General Public     *
 *   License along with this library; if not, write to the Free Software   *
 *   Foundation, Inc., 59 Temple Place - Suite 330, Boston,                *
 *   MA 02111-1307, USA                                                    *
 ***************************************************************************/


#include <cstdlib>
#include <cstring>
#include <iostream>
#include <sstream>
#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/thread/thread.hpp>
#include <vector>

#include <boost/algorithm/string/trim.hpp>

#include <boost/program_options.hpp>
namespace po = boost::program_options;

#include <boost/date_time/posix_time/posix_time.hpp>
using namespace boost::posix_time;


#include "bocat/common/base64.hpp"
#include "bocat/common/logging.hpp"

#include "sakcab/packet.pb.h"


using boost::asio::ip::tcp;

enum { max_length = 1024 };

#define N_ROUND 100000


static bocat::common::Logger*   logger_p = NULL;

#define INFO(ARGS...)   logger_p->info(__LINE__,__FILE__,ARGS)
#define ERROR(ARGS...)  logger_p->error(__LINE__,__FILE__,ARGS)

float
getTimeDuration( ptime time_begin, ptime time_end )
{
    time_period tp( time_begin, time_end );
    time_duration   td = tp.length();
    long    time_last_ms = td.total_milliseconds();
    float   time_last_s = time_last_ms / 1000.0f;

    return time_last_s;
}


void    ping(const char* host, const char* port, int times, bool verbose)
{
	try
	{
		boost::asio::io_service io_service;

		INFO(" * ping service %s:%s ...", host, port);
		tcp::resolver resolver(io_service);
		//tcp::resolver::query query(tcp::v4(), host, port);
		tcp::resolver::query query(host, port);
		tcp::resolver::iterator iterator = resolver.resolve(query);

		tcp::socket s(io_service);
		s.connect(*iterator);

        ptime   time_start = microsec_clock::local_time();
        ptime   time_start_R = time_start;
		u_int32_t   count_down_R = N_ROUND;
		u_int32_t   count_round = 0;

		for (int i=0; i<times; i++) {
			/*
			char request[max_length];
			sprintf(request, "hello-world-%09d\r\n", i);
			size_t request_length = strlen(request);
			boost::asio::write(s, boost::asio::buffer(request, request_length));
			*/

			sakcab::Packet	pkt_request;
			pkt_request.set_sid( i + 1000 );
			pkt_request.set_type( sakcab::Packet::REQUEST );
			pkt_request.set_op( sakcab::Packet::PING );

			sakcab::Packet::Item*   item_p = NULL;
			
			item_p = pkt_request.add_item();
			item_p->set_table( "test table" );
			item_p->set_key( "key001" );
			item_p->set_value( "value001" );

			item_p = pkt_request.add_item();
			item_p->set_kpin( "key pin" );
			item_p->set_key( "key002" );
			item_p->set_value( "value002" );
			
			std::string	pkt_str;
			if ( !pkt_request.SerializeToString( &pkt_str ) ) {
				throw std::runtime_error("fail to serialize packet to string.");
			}

			std::ostringstream	request_stream;
			request_stream << bocat::common::base64_encode(pkt_str) << "\r\n";
			std::string	request = request_stream.str();
			/*
			{
				std::string	request_test = request;
				boost::trim( request_test );
				sakcab::Packet	pkt_request_test;
				if (!pkt_request_test.ParseFromString( request_test ) ) {
					throw std::runtime_error("fail to test parsing Packet from string.");
				}
			}
			*/
			boost::asio::write(s, boost::asio::buffer(request.c_str(), request.size()) );

			//if (verbose) {
			//	std::cout << "REQUEST : " << request << std::endl;
			//}

			boost::asio::streambuf response;
			boost::asio::read_until(s, response, "\r\n");

			std::istream _stream( &response );
			std::string	response_str;
			std::getline( _stream, response_str );
			boost::trim( response_str );

			if (verbose) {
				std::cout << "RESPONSE: " << response_str << std::endl;
			}

			std::string	response_plain = bocat::common::base64_decode< std::string >( response_str );
			sakcab::Packet	pkt_response;
			pkt_response.ParseFromString( response_plain );

			if ( 1 != pkt_response.sid() ) {
				throw std::runtime_error( "response packet assert (sid == 1)" );
			}
			if ( pkt_response.reqid() != pkt_request.sid() ) {
				std::ostringstream	msg;
				msg << "BAD reqid ( want:" << pkt_request.sid() << " got:" << pkt_response.reqid() << " )";
				throw std::runtime_error( msg.str() );
			}
			if ( 2 != pkt_response.item_size() ) {
				throw std::runtime_error( "response packet assert (item_size() == 2)" );
			}

            --count_down_R;
            if (0 == count_down_R) {
                ptime   time_stop_R = microsec_clock::local_time();
                float   time_last_s = getTimeDuration( time_start_R, time_stop_R );
                float   tps = N_ROUND / time_last_s;
                INFO("  R-PING %u*%u records last %.3f seconds, tps=%.2f", ++count_round, N_ROUND, time_last_s, tps);

                time_start_R = time_stop_R;
                count_down_R = N_ROUND;
            }
		}

        ptime   time_stop = microsec_clock::local_time();
        float   time_last_s = getTimeDuration( time_start, time_stop );
        float   tps = times / time_last_s;
        INFO("    PING %u times last %.3f seconds, tps=%.2f", times, time_last_s, tps);
 
	}
	catch (std::exception& e)
	{
		std::cerr << "Exception: " << e.what() << "\n";
	}
}


int main(int argc, char* argv[])
{

	std::string	host;
	std::string	port;
	int times = 0;
	int thread_count = 0;
	bool verbose = false;

    try {

        po::options_description desc("Allowed options");
        desc.add_options()
            ("help,h", "produce help message")
            ("verbose,v", "output verbose message")
            ("host", po::value<std::string>(&host)->default_value("0.0.0.0"), "server ip/hostname")
            ("port", po::value<std::string>(&port)->default_value("5137"), "server port")
            ("times", po::value<int>(&times)->default_value(10), "number of request to send")
            ("threads", po::value<int>(&thread_count)->default_value(1), "number of threads to run")
        ;   

        po::variables_map vm;    
        po::store(po::parse_command_line(argc, argv, desc), vm);
        po::notify(vm);    

        if (vm.count("help")) {
            std::cout << desc << "\n";
            return 1;
        }   

		if (vm.count("verbose")) {
			verbose = true;
		}

		std::cout << "  server : " << host << std::endl
				  << "    port : " << port << std::endl
				  << "   times : " << times << std::endl
				  << " threads : " << thread_count << std::endl
				  ;
    }   
    catch(std::exception& e) {
        std::cerr << "error: " << e.what() << "\n";
        return 1;
    }   
    catch(...) {
        std::cerr << "Exception of unknown type!\n";
		return 2;
    }   

	GOOGLE_PROTOBUF_VERIFY_VERSION;

	logger_p = bocat::common::getLogger();

	std::vector< boost::thread * >    threads;

	for (int i=0; i<thread_count; i++) {
		boost::thread*   thr = new boost::thread( boost::bind(ping, host.c_str(), port.c_str(), times, verbose) );

		threads.push_back( thr );
	}

	for (size_t i=0; i<threads.size(); i++) {
		boost::thread*  thr = threads[ i ];
		thr->join(); 
	}

	for (size_t i=0; i<threads.size(); i++) {
		boost::thread*  thr = threads[ i ];
		delete thr;
	}
	threads.clear();

	google::protobuf::ShutdownProtobufLibrary();

	return 0;
}

